/*
 *
 *  * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 *
 *
 */

package com.iwindplus.base.http.client;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.iwindplus.base.http.client.domain.property.WebClientProperty;
import com.iwindplus.base.http.client.domain.property.WebClientProperty.Pool;
import com.iwindplus.base.http.client.domain.property.WebClientProperty.Pool.PoolType;
import io.micrometer.observation.ObservationRegistry;
import io.netty.channel.ChannelOption;
import io.netty.handler.codec.http.HttpRequestDecoder;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutException;
import io.netty.handler.timeout.WriteTimeoutHandler;
import jakarta.annotation.Resource;
import java.net.ConnectException;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.web.ServerProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.cloud.client.loadbalancer.LoadBalanced;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import org.springframework.http.codec.ClientCodecConfigurer.ClientDefaultCodecs;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.ExchangeStrategies;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Mono;
import reactor.netty.http.HttpProtocol;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.client.HttpResponseDecoderSpec;
import reactor.netty.resources.ConnectionProvider;
import reactor.netty.transport.ProxyProvider;
import reactor.util.retry.Retry;

/**
 * WebClient配置.
 *
 * @author zengdegui
 * @since 2025/08/24 01:38
 */
@Slf4j
@Configuration
@EnableConfigurationProperties({WebClientProperty.class})
@ConditionalOnProperty(prefix = "web.client", name = "enabled", havingValue = "true", matchIfMissing = true)
public class WebClientConfiguration {

    @Resource
    private ServerProperties serverProperties;

    @Resource
    private WebClientProperty property;

    @Resource
    private ObjectMapper objectMapper;

    @Resource
    private ObservationRegistry observationRegistry;

    /**
     * 创建HttpClient实例.
     *
     * @return HttpClient
     */
    @Bean
    @ConditionalOnMissingBean(HttpClient.class)
    public HttpClient webClientHttpClient() {
        HttpClient httpClient =
            HttpClient.create(this.buildConnectionPool())
                .httpResponseDecoder(this::httpResponseDecoder)
                .doOnConnected(
                    conn ->
                        conn.addHandlerLast(
                            new HttpRequestDecoder(
                                property.getMaxInitialLineLength(),
                                property.getMaxHeaderSize(),
                                property.getMaxChunkSize())))
                .doOnConnected(
                    c ->
                        c.addHandlerLast(
                                new ReadTimeoutHandler(
                                    property.getReadTimeout().getSeconds(), TimeUnit.SECONDS))
                            .addHandlerLast(
                                new WriteTimeoutHandler(
                                    property.getWriteTimeout().getSeconds(), TimeUnit.SECONDS)));
        if (serverProperties.getHttp2().isEnabled()) {
            httpClient = httpClient.protocol(HttpProtocol.HTTP11, HttpProtocol.H2);
        }
        if (property.getConnectTimeout() != null) {
            httpClient =
                httpClient.option(
                    ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) property.getConnectTimeout().toMillis());
        }
        if (property.getResponseTimeout() != null) {
            httpClient = httpClient.responseTimeout(property.getResponseTimeout());
        }
        if (property.getEnabledSoSndBuf()) {
            httpClient = httpClient.option(ChannelOption.SO_SNDBUF, property.getSoSndBuf());
        }
        if (property.getEnabledWiretap()) {
            httpClient = httpClient.wiretap(true);
        }
        if (property.getEnabledCompression()) {
            httpClient = httpClient.compress(true);
        }
        httpClient = this.configureProxy(httpClient);
        log.info("HttpClient={}", httpClient);
        return httpClient;
    }

    /**
     * 创建ExchangeStrategies实例.
     *
     * @return ExchangeStrategies
     */
    @Bean
    public ExchangeStrategies webClientExchangeStrategies() {
        final ExchangeStrategies exchangeStrategies =
            ExchangeStrategies.builder()
                .codecs(configurer -> {
                    final ClientDefaultCodecs clientDefaultCodecs = configurer.defaultCodecs();
                    clientDefaultCodecs.jackson2JsonDecoder(new Jackson2JsonDecoder(objectMapper));
                    clientDefaultCodecs.jackson2JsonEncoder(new Jackson2JsonEncoder(objectMapper));
                    clientDefaultCodecs.maxInMemorySize(property.getMaxInMemorySize());
                    clientDefaultCodecs.enableLoggingRequestDetails(property.getEnableLoggingRequestDetails());
                }).build();
        log.info("ExchangeStrategies={}", exchangeStrategies);
        return exchangeStrategies;
    }

    /**
     * 创建负载均衡 WebClient.Builder.
     *
     * @param strategies 策略
     * @param httpClient http客户端
     * @return WebClient.Builder
     */
    @LoadBalanced
    @Bean("loadBalancedWebClientBuilder")
    public WebClient.Builder loadBalancedWebClientBuilder(ExchangeStrategies strategies, HttpClient httpClient) {
        final WebClient.Builder webClientBuilder = createWebClientBuilder(strategies, httpClient);
        log.info("LoadBalanced WebClient.Builder={}", webClientBuilder);
        return webClientBuilder;
    }

    /**
     * 创建 WebClient.Builder.
     *
     * @param strategies 策略
     * @param httpClient http客户端
     * @return WebClient
     */
    @Bean
    public WebClient.Builder webClientBuilder(ExchangeStrategies strategies, HttpClient httpClient) {
        final WebClient.Builder webClientBuilder = createWebClientBuilder(strategies, httpClient);
        log.info("WebClient.Builder={}", webClientBuilder);
        return webClientBuilder;
    }

    /**
     * 创建负载均衡 WebClient.
     *
     * @param loadBalancedWebClientBuilder loadBalancedWebClientBuilder
     * @return WebClient
     */
    @Bean("loadBalancedWebClient")
    public WebClient loadBalancedWebClient(WebClient.Builder loadBalancedWebClientBuilder) {
        final WebClient webClient = loadBalancedWebClientBuilder.build();
        log.info("LoadBalanced WebClient={}", webClient);
        return webClient;
    }

    /**
     * 创建 WebClient.
     *
     * @param webClientBuilder webClientBuilder
     * @return WebClient
     */
    @Primary
    @Bean
    public WebClient webClient(WebClient.Builder webClientBuilder) {
        final WebClient webClient = webClientBuilder.build();
        log.info("WebClient={}", webClient);
        return webClient;
    }

    private WebClient.Builder createWebClientBuilder(ExchangeStrategies strategies, HttpClient httpClient) {
        WebClient.Builder builder = WebClient.builder()
            .exchangeStrategies(strategies)
            .clientConnector(new ReactorClientHttpConnector(httpClient))
            .filter(logHandler())
            .filter(retryHandler());

        if (Boolean.TRUE.equals(property.getEnabledObservation())) {
            builder.observationRegistry(observationRegistry);
        }
        return builder;
    }

    private ConnectionProvider buildConnectionPool() {
        final Pool pool = this.property.getPool();

        ConnectionProvider connectionProvider;
        if (pool.getType() == PoolType.DISABLED) {
            connectionProvider = ConnectionProvider.newConnection();
        } else {
            ConnectionProvider.Builder builder = ConnectionProvider.builder(pool.getName());
            if (pool.getType() == PoolType.FIXED) {
                builder
                    .maxConnections(pool.getMaxConnections())
                    .pendingAcquireMaxCount(-1)
                    .pendingAcquireTimeout(pool.getPendingAcquireTimeout());
            } else {
                builder
                    .maxConnections(Integer.MAX_VALUE)
                    .pendingAcquireMaxCount(-1)
                    .pendingAcquireTimeout(Duration.ofMillis(0));
            }
            if (pool.getMaxIdleTime() != null) {
                builder.maxIdleTime(pool.getMaxIdleTime());
            }
            if (pool.getMaxLifeTime() != null) {
                builder.maxLifeTime(pool.getMaxLifeTime());
            }
            builder.evictInBackground(pool.getEvictionInterval());
            builder.metrics(pool.getMetrics());
            connectionProvider = builder.build();
        }
        log.info("ConnectionProvider={}", connectionProvider);
        return connectionProvider;
    }

    private HttpResponseDecoderSpec httpResponseDecoder(HttpResponseDecoderSpec spec) {
        spec.maxHeaderSize(property.getMaxHeaderSize())
            .maxInitialLineLength(property.getMaxInitialLineLength());
        return spec;
    }

    private HttpClient configureProxy(HttpClient client) {
        final WebClientProperty.Proxy proxy = property.getProxy();
        if (Boolean.FALSE.equals(proxy.getEnabled())) {
            return client;
        }

        return client.proxy(
            ps -> {
                ProxyProvider.Builder b = ps.type(proxy.getType()).host(proxy.getHost());
                PropertyMapper m = PropertyMapper.get();
                m.from(proxy::getPort).whenNonNull().to(b::port);
                m.from(proxy::getUsername).whenHasText().to(b::username);
                m.from(proxy::getPassword).whenHasText().to(pw -> b.password(s -> pw));
                m.from(proxy::getNonProxyHostsPattern).whenHasText().to(b::nonProxyHosts);
            });
    }

    private ExchangeFilterFunction logHandler() {
        if (Boolean.FALSE.equals(property.getEnabledRequestLog())) {
            return ExchangeFilterFunction.ofRequestProcessor(Mono::just);
        }

        return (request, next) ->
            Mono.deferContextual(
                ctx -> {
                    long start = System.nanoTime();
                    return next.exchange(request)
                        .doOnNext(
                            resp ->
                                log.info(
                                    "{} {} -> {} (cost={}ms)",
                                    request.method(),
                                    request.url(),
                                    resp.statusCode(),
                                    TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start)))
                        .doOnError(
                            err ->
                                log.error(
                                    "{} {} -> ERROR: {} (cost={}ms)",
                                    request.method(),
                                    request.url(),
                                    err.getClass().getSimpleName(),
                                    TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start), err));
                });
    }

    private ExchangeFilterFunction retryHandler() {
        final WebClientProperty.Retry retry = property.getRetry();
        if (Boolean.FALSE.equals(retry)) {
            return ExchangeFilterFunction.ofRequestProcessor(Mono::just);
        }

        return (request, next) ->
            next.exchange(request)
                .timeout(property.getReadTimeout())
                .retryWhen(
                    Retry.backoff(retry.getMaxAttempts(), retry.getPeriod())
                        .maxBackoff(retry.getMaxBackoffTimeout())
                        .transientErrors(true)
                        .filter(this::isRetryable));
    }

    /**
     * 判定是否可重试：仅网络、IO、超时异常.
     */
    private boolean isRetryable(Throwable ex) {
        return ex instanceof ReadTimeoutException
            || ex instanceof WriteTimeoutException
            || ex instanceof ConnectException;
    }
}
